#include "signaling_server.h"
#include "signaling_worker.h"
#include "base/socket.h"
#include "rtc_base/logging.h"

#include <yaml-cpp/yaml.h>
#include "rtc_server.h"


namespace xrtc {

// 管道被写入的回调处理    
void signalingServerRecvNotify(EventLoop* /*el*/, IOWatcher* /*w*/, int fd, int /*events*/, void* data) {
    int msg;
    if (read(fd, &msg, sizeof(int)) != sizeof(int)) {
        RTC_LOG(LS_WARNING) << "read from pipe error: " << strerror(errno) << ", errno: " << errno;
        return;
    }

    SignalingServer* server = (SignalingServer*)data;
    server->_processNotify(msg);
}


// 有新连接时的回调处理
void acceptNewConn(EventLoop* /* el */, IOWatcher* /* w */, int fd, int /* events */, void *data) {
    int cfd;
    char cip[128];
    int cport;

    cfd = tcp_accept(fd, cip, &cport);
    if (cfd == -1) {
        return;
    }

    RTC_LOG(LS_INFO) << "accept new conn, fd: " << cfd << ", ip: " << cip << ", port: " << cport;

    SignalingServer* server = (SignalingServer*)data;
    server->_dispatch_new_conn(cfd);
}

// 处理写入管道中的数据
void SignalingServer::_processNotify(int msg) {
    switch (msg) {
        case QUIT:
            _stop();
            break;
        default:
            RTC_LOG(LS_WARNING) << "unknown msg: " << msg;
            break;
    }
}


// 停止事件循环
void SignalingServer::_stop()
{
    if (!m_thread) {
        RTC_LOG(LS_WARNING) << "signaling server not running";
        return;
    }

    m_el->deleteIOEvent(m_pipe_watcher);
    m_el->deleteIOEvent(m_io_watcher);
    m_el->stop();

    close(m_notify_recv_fd);
    close(m_notify_send_fd);
    close(m_listen_fd);

    RTC_LOG(LS_INFO) << "signaling server stop";

    for (auto worker : m_workers) {
        if (worker) {
            worker->stop();
            worker->join();
        }
    }
}


// 创建工作线程
int SignalingServer::_createWorker(int worker_id)
{
    RTC_LOG(LS_INFO) << "SignalingServer create worker, worker_id: " << worker_id;
    SignalingWorker* worker = new SignalingWorker(worker_id, m_options);

    if (worker->init() != 0) {
        return -1;
    }

    if (!worker->start()) {
        return -1;
    }

    m_workers.push_back(worker);

    return 0;
}

// 分发新连接
void SignalingServer::_dispatch_new_conn(int fd) {
    int index = m_next_worker_index;
    m_next_worker_index = (m_next_worker_index + 1) % m_options.m_worker_num;

    SignalingWorker* worker = m_workers[index];
    // 发送通知给对应的线程，新连接到来
    worker->notifyNewConn(fd);
}


SignalingServer::SignalingServer() : m_el(new EventLoop(this)) {

}


SignalingServer::~SignalingServer()
{
    if (m_el) {
        delete m_el;
        m_el = nullptr;
    }

    if (m_thread) {
        m_thread.reset();
    }

    for (auto worker : m_workers) {
        if (worker) {
            delete worker;
            worker = nullptr;
        }
    }

    m_workers.clear();
}


int SignalingServer::init(const char *conf_file)
{
    if (conf_file == nullptr) {
        RTC_LOG(LS_WARNING) << "signaling server conf_file is null";
        return -1; 
    }

    try {
        YAML::Node config = YAML::LoadFile(conf_file);
        RTC_LOG(LS_INFO) << "signaling server options:\n" << config;
        
        m_options.m_host = config["host"].as<std::string>();
        m_options.m_port = config["port"].as<int>();
        m_options.m_worker_num = config["worker_num"].as<int>();
        m_options.m_connection_timeout = config["connection_timeout"].as<int>();

    } catch (const YAML::Exception& e) {
        RTC_LOG(LS_WARNING) << "catch a YAML exception, line:" << e.mark.line + 1
            << ", column: " << e.mark.column + 1 << ", error: " << e.msg;
        return -1;
    }

    // 创建管道用于线程间的通信
    int pipefd[2];
    if (pipe(pipefd)) {
        RTC_LOG(LS_WARNING) << "create pipe error: " << strerror(errno) << ", errno: " << errno;
        return -1;
    }

    m_notify_recv_fd = pipefd[0];
    m_notify_send_fd = pipefd[1];

    // 将recv_fd添加到事件循环中
    m_pipe_watcher = m_el->createIOEvent(signalingServerRecvNotify, this);
    m_el->startIOEvent(m_pipe_watcher, m_notify_recv_fd, EventLoop::READ);

    // 创建tcp server
    m_listen_fd = create_tcp_server(m_options.m_host.c_str(), m_options.m_port); 
    if (m_listen_fd == -1) {
        return -1;
    }

    m_io_watcher = m_el->createIOEvent(acceptNewConn, this);
    m_el->startIOEvent(m_io_watcher, m_listen_fd, EventLoop::READ);

    // 创建多个worker线程
    for (int i = 0; i < m_options.m_worker_num; i++) {
         if (_createWorker(i) != 0) {
            return -1;
        }
    }

    return 0;
}


// 创建工作线程并开启事件循环
bool SignalingServer::start() {
    if (m_thread) {
        RTC_LOG(LS_WARNING) << "signaling server already start";
        return false;
    }

    m_thread = std::make_unique<std::thread>([=]() {
        RTC_LOG(LS_INFO) << "signaling server event loop run";
        m_el->start();
        RTC_LOG(LS_INFO) << "signaling server event loop quit";
    });

    return true;
}

// 向管道发送停止事件循环的信号（由其它线程调用）
void SignalingServer::stop() {
    notify(SignalingServer::QUIT);
}

// 向管道中写入数据
int SignalingServer::notify(int msg) {
    int written_len = write(m_notify_send_fd, &msg, sizeof(int));

    return written_len == sizeof(int) ? 0 : -1;
}


// 等待工作线程结束
void SignalingServer::join() {
    if (m_thread && m_thread->joinable()) {
        m_thread->join();
    }
}

} // namespace xrtc